package com.zifeng.service.impl;
import java.util.Date;

import com.alibaba.fastjson.JSON;
import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
import com.baomidou.mybatisplus.core.metadata.IPage;
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import com.zifeng.common.Query;
import com.zifeng.common.R;
import com.zifeng.dao.ShipDao;
import com.zifeng.dao.UserDao;
import com.zifeng.dao.UserPraiseDao;
import com.zifeng.exception.BizException;
import com.zifeng.feign.BlockFeignClient;
import com.zifeng.feign.OrderFeignClient;
import com.zifeng.pojo.*;
import com.zifeng.rocketmq.OrderRocketMQProducer;
import com.zifeng.service.IShipService;
import com.zifeng.utils.BloomFilterUtil;
import com.zifeng.utils.PageUtils;
import com.zifeng.utils.RedisUtils;
import jakarta.annotation.PostConstruct;
import jakarta.annotation.Resource;
import lombok.extern.slf4j.Slf4j;
import org.apache.seata.spring.annotation.GlobalTransactional;
import org.redisson.api.RBloomFilter;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import java.util.*;

import static com.zifeng.constant.RedisConstants.*;
import static com.zifeng.exception.ErrorEnum.USER_ACCOUNT_MONEY_NOT_ENOUGH;

/**
 * 船舶船只Service业务层处理
 * 
 * @author zifeng
 * @date 2025-03-27
 */
@Slf4j
@Service("shipService")
public class ShipServiceImpl extends ServiceImpl<ShipDao, Ship> implements IShipService {



    @Autowired
    private UserPraiseDao userPraiseDao;

    @Autowired
    private ShipDao shipDao;

    @Autowired
    private UserDao userDao;

    @Resource
    private OrderFeignClient orderFeignClient;

    @Resource
    private BlockFeignClient blockFeignClient;

    @Autowired
    private RedisUtils redisUtils;

    @Autowired
    private BloomFilterUtil bloomFilterUtil;

    private RBloomFilter<Long> bloomFilter = null;

    @Autowired
    private OrderRocketMQProducer orderRocketMQProducer;

    @PostConstruct
    public void init() {
        List<Ship> list = this.list();
        bloomFilter = bloomFilterUtil.create(SHIP_ID_BLOOM_FILTER, EXPECTED_INSERTIONS, FALSE_PROBABILITY);
        for (Ship ship : list) {
            bloomFilter.add(ship.getId());
        }
    }

    @Override
    public PageUtils queryPage(Map<String, Object> params) {
        QueryWrapper<Ship> wrapper = new QueryWrapper<>();
        Object ownerId = params.get("ownerId");
        if(ownerId != null) {
            int newOwnerId = Integer.parseInt((String) ownerId);
            wrapper.eq("owner_id", newOwnerId);
        }
        wrapper.orderByDesc("publish_year");
        IPage<Ship> page = this.page(
                new Query<Ship>().getPage(params),
                wrapper
        );

        return new PageUtils(page);
    }

    @Override
    public PageUtils queryPageByUuserPraise(Map<String, Object> params) {

        Object userId = params.get("userId");

        int newUserId = Integer.parseInt((String) userId);
        QueryWrapper<UserPraise> wrapper = new QueryWrapper<>();
        wrapper.eq("user_id", newUserId);
        List<UserPraise> userPraises = userPraiseDao.selectList(wrapper);
        List<Integer> shipIdList = new ArrayList<>();
        for (UserPraise userPraise : userPraises) {
            shipIdList.add(userPraise.getShipId());
        }
        int pageSize = Integer.parseInt(params.get("limit").toString());
        int currPage = Integer.parseInt(params.get("page").toString());
        if(shipIdList.size() > 0) {
            List<Ship> shipList = this.listByIds(shipIdList);
            return new PageUtils(shipList, shipList.size(), pageSize, currPage);
        }else {
            return new PageUtils(null, 0, pageSize, currPage);
        }
    }


    @Override
    public R buyShip(ShipDto shipDto) {
        // 1、检查买家账户余额是否足够
        int shipId = shipDto.getShipId();
        Ship ship = shipDao.selectById(shipId);

        User buyer = userDao.selectById(shipDto.getBuyerId());
        long buyerRemainMoney = buyer.getMoney() - ship.getPrice();
        if(buyerRemainMoney < 0) {
            throw new BizException(USER_ACCOUNT_MONEY_NOT_ENOUGH);
        }

        // 2、给RocketMQ发送事务消息
        Order order = new Order();
        order.setMasterId(shipDto.getSellerId());
        order.setClientId(shipDto.getBuyerId());
        order.setShipId(shipDto.getShipId());
        order.setStatus(0);
        order.setPrice(ship.getPrice());
        order.setCreateAt(new Date());
        orderRocketMQProducer.sendTransactionMessage("order-transaction-topic", order, shipDto);

        return R.ok();
    }

    // 执行RocketMQ本地事务对应业务操作
    @GlobalTransactional
    public void doMysqlLocalTransaction(Order order, ShipDto shipDto) {
        // 1. 查询船只信息
        int shipId = shipDto.getShipId();
        Ship ship = shipDao.selectById(shipId);

        // 2. 远程调用创建订单
        orderFeignClient.addOrderWithBlock(order);

        // 将订单添加到区块链中
        // 因为Kafka的事务消息不合需求，所以采用RocketMQ的方式发送消息

//        R mine = blockFeignClient.mine(order);
//        if(mine.get("error") != null) {
//            throw (BizException) mine.get("error");
//        }

        // 3. 扣除买家账户余额，增加卖家账户金额
        User buyer = userDao.selectById(shipDto.getBuyerId());
        long buyerRemainMoney = buyer.getMoney() - ship.getPrice();
        buyer.setMoney(buyerRemainMoney);
        userDao.updateById(buyer);
        User seller = userDao.selectById(shipDto.getSellerId());
        long sellerRemainMoney = seller.getMoney() + ship.getPrice();
        seller.setMoney(sellerRemainMoney);
        userDao.updateById(seller);

        // 4. 修改船只归属
        ship.setOwnerId(buyer.getId());
        shipDao.updateById(ship);
    }

    /**
     * 查询船只
     * 
     * @param id 船只主键
     * @return 船只
     */
    @Override
    public Ship selectShipById(Long id) {
        // 使用布隆过滤器判断是否存在
        if(!bloomFilter.contains(id)) {
            log.info("Ship ID {} not found in bloom filter, returning null", id);
            redisUtils.set(SHIP_PREFIX + id, "null", SHIP_NULL_TTL, TTL_TIME_UNIT);
            return null;
        }
        String shipKey = SHIP_PREFIX + id;
        Ship ship = JSON.parseObject(redisUtils.get(shipKey), Ship.class);
        if(ship != null) return ship;
        ship = shipDao.selectById(id);
        redisUtils.set(shipKey, JSON.toJSONString(ship), SHIP_TTL, TTL_TIME_UNIT);
        return ship;
    }

    /**
     * 新增船只
     * 
     * @param ship 船只
     * @return 结果
     */
    @Override
    public int insertShip(Ship ship)
    {
        return shipDao.insert(ship);
    }

    /**
     * 修改船只
     * 
     * @param ship 船只
     * @return 结果
     */
    @Override
    public int updateShip(Ship ship)
    {
        return shipDao.updateById(ship);
    }

    /**
     * 批量删除船只
     * 
     * @param ids 需要删除的船只主键
     * @return 结果
     */
    @Override
    public int deleteShipByIds(Long[] ids)
    {
        return shipDao.deleteBatchIds(Arrays.asList(ids));
    }

    /**
     * 删除船只信息
     * 
     * @param id 船只主键
     * @return 结果
     */
    @Override
    public int deleteShipById(Long id)
    {
        return shipDao.deleteById(id);
    }
}
